package cn.josewu.util

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.streaming.kafka010.OffsetRange


/**
 * @program: Covid19_WZData_process
 * @description: 用来手动维护偏移量到MYSQL中的工具类
 * @author: Mr. Tigger
 * @create 2020-11-14 09:11
 **/

object OffsetUtils {


  def saveOffsets(groupid: String, offsets: Array[OffsetRange]): Unit = {
       //加载驱动并获取连接
       val conn:Connection =DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8","root","13456")
       //编写SQL
       val sql:String = "replace into t_offset(topic,partition,groupid,offset) values(?,?,?,?)"
       //创建预编译语句对象
       val ps: PreparedStatement = conn.prepareStatement(sql)
       //设置参数并执行
       for (o<- offsets) {
         ps.setString(  1,o.topic)
         ps.setInt(2,o.partition)
         ps.setString(3,groupid)
         ps.setLong(4,o.untilOffset)
         ps.executeLargeUpdate()
       }
       //关闭资源
  }

}
